package com.etlmaster.executor.bigdata.hbase;

import java.io.File;
import java.net.URI;
import java.util.UUID;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;

import com.etlmaster.executor.utils.LogWriter;


/**
 * @author Wugf
 * @date 2016-11-10
 */
public class HbaseBulkLoad {
	private String uri;
	private HTable htable;

	private Configuration conf;

	private Path outputPath;
	private String user;
	private String tableName;
	private String outPutPath;
	private String columnFamilys;
	private String fileFields;
	private String rowKeyColumns;
	private String seprator;
	public HbaseBulkLoad(String user,String tableName,String inputPath,String outPutPath,String columnFamilys,String fileFields,String rowKeyColumns,String seprator){
		this.user=user;
		this.tableName=tableName;
		this.uri=inputPath;
		this.outPutPath=outPutPath;
		this.columnFamilys=columnFamilys;
		this.fileFields=fileFields;
		this.rowKeyColumns=rowKeyColumns;
		this.seprator=seprator;
	}
	public static void main(String []args){
		new HbaseBulkLoad("lighre","test","/test","/test_out","col1,col2,col3,col4,col5,col6","col1,col2,col3,col4,col5,col6","col1,col2",",").bulkLoad();
	}
	public void initConf(){
		Configuration conf = HBaseConfiguration.create();
		File coreSitefile=new File(Environment.CORE_SITE);
		File hdfsSitefile=new File(Environment.HDFS_SITE);
		File hbaseSitefile=new File(Environment.HBSE_SITE);
		File yarnSitefile=new File(Environment.YARN_SITE);
		if(coreSitefile.exists()&&hdfsSitefile.exists()&&hbaseSitefile.exists()&&yarnSitefile.exists())
		{
			Path coreSitepath=new Path(Environment.CORE_SITE);
			Path hdfsSitepath=new Path(Environment.HDFS_SITE);
			Path hbaseSitePath=new Path(Environment.HBSE_SITE);

			Path yarnSitePath=new Path(Environment.YARN_SITE);
			conf.addResource(coreSitepath);
			conf.addResource(hdfsSitepath);
			conf.addResource(hbaseSitePath);
			conf.addResource(yarnSitePath);
		}
		System.setProperty("HADOOP_USER_NAME", user);
		conf.set("fileFieldName", fileFields);// 文件字段
		conf.set("fileSeparator",seprator);// 文件分隔符
		conf.set("columnFamily", columnFamilys);
		conf.set("rowKeyColumn", rowKeyColumns);
        conf.set("mapreduce.framework.name", "yarn");  
        conf.setBoolean("mapred.compress.map.output", true);  
        conf.setClass("mapred.map.output.compression.codec",GzipCodec.class, CompressionCodec.class);  

        this.conf = conf;

	}
	public boolean bulkLoad()
	{
		try
		{
			initConf();
			if (hfileGenerator(this.conf))
			{
				LogWriter.addLog("INFO","开始执行bulkload,{}",htable.getName().toString());
				LoadIncrementalHFiles loader = new LoadIncrementalHFiles(this.conf);
				loader.doBulkLoad(outputPath, htable);
				LogWriter.addLog("INFO","bulkload执行结束");
				
				LogWriter.addLog("INFO","hbase装载成功！！！");
			}
			else
			{
				
				LogWriter.addLog("INFO","hbase装载失败！！！，mapreduce执行出错！！！");
				return false;
			}
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","bulkload失败！！！");
			e.printStackTrace();
			
			return false;
		}
		return true;
	}
	private boolean hfileGenerator(Configuration conf)
	{
		boolean success = true;
		try
		{
			this.htable = new HTable(conf, tableName);
	        String jobName = tableName + UUID.randomUUID();
			Job job = Job.getInstance(conf, jobName);
			job.setJarByClass(HbaseBulkLoad.class);

			FileSystem fs = FileSystem.get(URI.create(uri), conf);
			// 如果输出路径存在就先删掉，因为不允许输出路径事先存在
			Path outPath = new Path(outPutPath);
			if (fs.exists(outPath))
			{
				fs.delete(outPath, true);
				LogWriter.addLog("INFO","HDFS中输出路径已经存在，已经删除！！！");
			}
			
			Path[] inputPaths = {new Path(uri)};
			outputPath=new Path(outPutPath);
			FileInputFormat.setInputPaths(job, inputPaths);
			job.setInputFormatClass(TextInputFormat.class);
			
			FileOutputFormat.setOutputPath(job, outputPath);
			// 设置Mapper与Reducer类
			job.setMapperClass(HFileGenerateMapper.class);
			job.setReducerClass(KeyValueSortReducer.class);
			// 设置Map任务输出Key-Value类型,一定要为该类型,Value可以改为HBase的Put类型
			job.setOutputKeyClass(ImmutableBytesWritable.class);
			job.setOutputValueClass(KeyValue.class);
			job.setPartitionerClass(TotalOrderPartitioner.class);
			HFileOutputFormat2.configureIncrementalLoad(job, htable);
			TableMapReduceUtil.addDependencyJars(job);
			TableMapReduceUtil.addDependencyJars(job.getConfiguration(), new Class[] { StringUtils.class,HFileGenerateMapper.class});
			LogWriter.addLog("INFO","开始执行mapreduce生成hfile");
			success = job.waitForCompletion(true);
			LogWriter.addLog("INFO","mapreduce生成hfile执行完成，结果为[" + (success ? "成功" : "失败") + "]");
			
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","执行生成HFile的MapReduce任务出错！！！" );
			e.printStackTrace();
		}
		return success;
	}
}
